package com.mongdbtokuduswagger.service.impl;

import com.mongdbtokuduswagger.entity.*;
import com.mongdbtokuduswagger.service.MongodbSyncToKuduSwaggerService;
import com.mongdbtokuduswagger.service.SystemProcessLogService;
import com.mongdbtokuduswagger.utils.DateUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.OutputStream;
import java.net.URI;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Random;

/**
 * @author Daft_V
 */
@Service
@Slf4j
public class MongodbSyncKuduSwaggerServiceImpl implements MongodbSyncToKuduSwaggerService {

    @Resource(name = "primaryMongoTemplate")
    private MongoTemplate mongoTemplate;

    @Resource
    private SystemProcessLogService systemProcessLogService;

    @Value("${hdfs.hdfs_path1}")
    private String hdfsPath1;

    @Value("${hdfs.hdfs_path2}")
    private String hdfsPath2;

    private String hdfsPath = null;

    /**
     * MongoDb同步数据到Kudu
     *
     * @param swaggerTimeEntity 时间范围实体类
     * @throws ParseException
     */
    @Override
    public Boolean syncToKudu(SwaggerTimeEntity swaggerTimeEntity){
        try{
            //判断hdfs主用
            Configuration conf = new Configuration();
            Path[] paths = null;
            try {
                conf.set("fs.defaultFS", hdfsPath1);
                FileSystem hdfs = FileSystem.get(conf);
                FileStatus[] fs = hdfs.listStatus(new Path("/collectiondata/"));
                paths = FileUtil.stat2Paths(fs);
                hdfsPath = hdfsPath1;
            } catch (Exception e) {
                hdfsPath = hdfsPath2;
            }
            moToKueHighRemoteDetect(swaggerTimeEntity);
            moToKueHighRemotePulse(swaggerTimeEntity);
            moToKueHighRemoteSignal(swaggerTimeEntity);
            moToKueLowRemoteDetect(swaggerTimeEntity);
            moToKueLowRemotePulse(swaggerTimeEntity);
            moToKuLowRemoteSignal(swaggerTimeEntity);
            moToKuWeather(swaggerTimeEntity);
            moToKuUpTransLog(swaggerTimeEntity);
            moToKuBusiTransRunStatus(swaggerTimeEntity);
            moToKuSystemConnStatus(swaggerTimeEntity);
            moToKuUpdetectstatus(swaggerTimeEntity);
            moToKuBusiAlarmInfo(swaggerTimeEntity);
            moToKuAmbiTranAlarm(swaggerTimeEntity);
            moToKuUpTransmitterSecondData("upTransmitterSecondDataB/upTransmitterSecondDataB.txt","upTransmitterSecondDataDeptB");
            moToKuUpTransmitterSecondData("upTransmitterSecondDataC/upTransmitterSecondDataC.txt","upTransmitterSecondDataDeptC");
            moToKuUpTransmitterSecondData("upTransmitterSecondDataD/upTransmitterSecondDataD.txt","upTransmitterSecondDataDeptD");
            moToKuUpTransmitterSecondData("upTransmitterMinuteDataB/upTransmitterMinuteDataB.txt","upTransmitterMinuteDataDeptB");
            moToKuUpTransmitterSecondData("upTransmitterMinuteDataC/upTransmitterMinuteDataC.txt","upTransmitterMinuteDataDeptC");
            moToKuUpTransmitterSecondData("upTransmitterMinuteDataD/upTransmitterMinuteDataD.txt","upTransmitterMinuteDataDeptD");
            return true;
        }catch (Exception e){
            e.printStackTrace();
            return false;
        }

    }

    /**
     * 删除MongoDb中的数据
     */
    @Override
    public Boolean deleteMongodb(SwaggerTimeEntity swaggerTimeEntity) {
        try{
            String format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getEndTime()) + "000000";
            String start_format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getStartTime()) + "000000";
            Query query = new Query(Criteria.where("pid").gt(start_format).lt(format));
            // 高压遥测信息
            mongoTemplate.findAllAndRemove(query, HighRemoteDetect.class);
            // 高压电度信息
            mongoTemplate.findAllAndRemove(query, HighRemotePulse.class);
            // 高压遥信信息
            mongoTemplate.findAllAndRemove(query, HighRemoteSignal.class);
            // 低压遥测信息
            mongoTemplate.findAllAndRemove(query, LowRemoteDetect.class);
            // 低压电度信息
            mongoTemplate.findAllAndRemove(query, LowRemotePulse.class);
            // 低压遥信信息
            mongoTemplate.findAllAndRemove(query, LowRemoteSignal.class);
            // 天气数据
            Query weatherQuery=new Query();
            mongoTemplate.findAllAndRemove(weatherQuery, "weather_data");
            // 发射机日志信息
            mongoTemplate.findAllAndRemove(query, UpTransmitterLog.class);
            // 发射机连接状态
            Query busiTransRunQuery = new Query(Criteria.where("statuses.pid").lt(format));
            mongoTemplate.findAllAndRemove(busiTransRunQuery, BusiTransRun.class);
            // 各系统连接状态信息
            mongoTemplate.findAllAndRemove(query, SystemConnStatus.class);
            // 发射机状态信息秒数据
            mongoTemplate.findAllAndRemove(query, Updetectstatus.class);
            // 综合分析与报警信息
            mongoTemplate.findAllAndRemove(query, BusiAlarmInfo.class);
            // 警告信息
            String lastTime = DateUtil.getLastTime(2);
            Query moToKuAmbiTranAlarmQuery = new Query(Criteria.where("alarmTime").lt(lastTime));
            mongoTemplate.findAllAndRemove(moToKuAmbiTranAlarmQuery, AmbiTranAlarm.class);
            // 发射机秒数据
            Query moToKuUpTransmitterSecondDataQuery = new Query(Criteria.where("transmitters.pid").lt(format));
            mongoTemplate.findAllAndRemove(moToKuUpTransmitterSecondDataQuery, UpTransmitterSecondData.class,"upTransmitterSecondDataDeptB");
            mongoTemplate.findAllAndRemove(moToKuUpTransmitterSecondDataQuery, UpTransmitterSecondData.class,"upTransmitterSecondDataDeptC");
            mongoTemplate.findAllAndRemove(moToKuUpTransmitterSecondDataQuery, UpTransmitterSecondData.class,"upTransmitterSecondDataDeptD");
            mongoTemplate.findAllAndRemove(moToKuUpTransmitterSecondDataQuery, UpTransmitterSecondData.class,"upTransmitterMinuteDataDeptB");
            mongoTemplate.findAllAndRemove(moToKuUpTransmitterSecondDataQuery, UpTransmitterSecondData.class,"upTransmitterMinuteDataDeptC");
            mongoTemplate.findAllAndRemove(moToKuUpTransmitterSecondDataQuery, UpTransmitterSecondData.class,"upTransmitterMinuteDataDeptD");
            return true;
        }catch (Exception e){
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 高压遥测信息
     */
    public void moToKueHighRemoteDetect(SwaggerTimeEntity swaggerTimeEntity) {
        Configuration conf = new Configuration();
        conf.setBoolean("dfs.support.append", true);
        FileSystem fs = null;
        try {
            fs = FileSystem.get(URI.create(hdfsPath + "highRemoteDetect/highRemoteDetect.txt"), conf);
            OutputStream output = fs.create(new Path(hdfsPath + "highRemoteDetect/highRemoteDetect.txt"));
            String start_format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getStartTime()) + "000000";
            String format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getEndTime()) + "000000";
            Date startTime = new Date();
            Query query = new Query(Criteria.where("pid").gt(start_format).lt(format));
            List<HighRemoteDetect> highRemoteDetect = mongoTemplate.find(query, HighRemoteDetect.class);
            log.info("高压遥测数据开始写入hdfs");
            for (HighRemoteDetect remoteDetect : highRemoteDetect) {
                output.write((remoteDetect.getPid() + "," + remoteDetect.getRemotDetectId() + ","
                        + remoteDetect.getEValue() + "," + remoteDetect.getCreateDate()).getBytes("UTF-8"));
                output.write("\n".getBytes("UTF-8"));
            }
            log.info("高压遥测数据写入hdfs完成");
            Date endTime = new Date();
            systemProcessLogService.addSystemProcessLog("MongoDB_Kudu_接入方式", "ETL", startTime, endTime, highRemoteDetect.size(), highRemoteDetect.size(), "Y", "成功");
            log.info("高压遥测数据同步完成" + "即将开始清除mongdb数据");
            try {
                mongoTemplate.findAllAndRemove(query, HighRemoteDetect.class);
            } catch (Exception e) {
                log.error("高压遥测数据清除异常" + e.getMessage());
            }
            log.info("高压遥测数据清除完成");
            fs.close();
            output.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 高压电度信息
     * @param swaggerTimeEntity
     */
    public void moToKueHighRemotePulse(SwaggerTimeEntity swaggerTimeEntity) {
        Configuration conf = new Configuration();
        conf.setBoolean("dfs.support.append", true);
        FileSystem fs = null;
        try {
            fs = FileSystem.get(URI.create(hdfsPath + "highRemotePulse/highRemotePulse.txt"), conf);
            OutputStream output = fs.create(new Path(hdfsPath + "highRemotePulse/highRemotePulse.txt"));
            String start_format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getStartTime()) + "000000";
            String format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getEndTime()) + "000000";
            Date startTime = new Date();
            Query query = new Query(Criteria.where("pid").gt(start_format).lt(format));
            List<HighRemotePulse> highRemotePulse = mongoTemplate.find(query, HighRemotePulse.class);
            log.info("高压电度数据开始写入hdfs");
            for (HighRemotePulse remotePulse : highRemotePulse) {
                output.write((remotePulse.getPid() + "," + remotePulse.getRemotPulseId() + ","
                        + remotePulse.getEValue() + "," + remotePulse.getCreateDate()).getBytes("UTF-8"));
                output.write("\n".getBytes("UTF-8"));
            }
            log.info("高压电度数据写入hdfs完成");
            Date endTime = new Date();
            systemProcessLogService.addSystemProcessLog("MongoDB_Kudu_接入方式", "ETL", startTime, endTime, highRemotePulse.size(), highRemotePulse.size(), "Y", "成功");
            log.info("高压电度数据同步完成" + "即将开始清除mongdb数据");
            try {
                mongoTemplate.findAllAndRemove(query, HighRemotePulse.class);
            } catch (Exception e) {
                log.error("高压电度数据清除异常" + e.getMessage());
            }
            log.info("高压电度数据清除完成");
            fs.close();
            output.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 高压遥信信息
     * @param swaggerTimeEntity 时间范围实体类
     */
    public void moToKueHighRemoteSignal(SwaggerTimeEntity swaggerTimeEntity){
        Configuration conf = new Configuration();
        conf.setBoolean("dfs.support.append", true);
        FileSystem fs = null;
        try {
            fs = FileSystem.get(URI.create(hdfsPath+"highRemoteSignal/highRemoteSignal.txt"), conf);
            OutputStream output = fs.create(new Path(hdfsPath+"highRemoteSignal/highRemoteSignal.txt"));
            String start_format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getStartTime()) + "000000";
            String format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getEndTime()) + "000000";
            Date startTime = new Date();
            Query query = new Query(Criteria.where("pid").gt(start_format).lt(format));
            List<HighRemoteSignal> highRemoteSignal = mongoTemplate.find(query,HighRemoteSignal.class);
            log.info("高压遥信数据开始写入hdfs");
            for (HighRemoteSignal remoteSignal : highRemoteSignal) {
                output.write((remoteSignal.getPid()+","+remoteSignal.getRemoteSignalId()+","
                        +remoteSignal.getEValue()+"," +remoteSignal.getCreateDate()).getBytes("UTF-8"));
                output.write("\n".getBytes("UTF-8"));
            }
            log.info("高压遥信数据写入hdfs完成");
            Date endTime = new Date();
            systemProcessLogService.addSystemProcessLog("MongoDB_Kudu_接入方式", "ETL", startTime, endTime, highRemoteSignal.size(), highRemoteSignal.size(), "Y", "成功");
            log.info("高压遥信数据同步完成"+"即将开始清除mongdb数据");
            try {
                mongoTemplate.findAllAndRemove(query, HighRemoteSignal.class);
            } catch (Exception e) {
                log.error("高压遥信数据清除异常" + e.getMessage());
            }
            log.info("高压遥信数据清除完成");
            fs.close();
            output.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 低压遥测信息
     * @param swaggerTimeEntity
     */
    public void moToKueLowRemoteDetect(SwaggerTimeEntity swaggerTimeEntity){
        Configuration conf = new Configuration();
        conf.setBoolean("dfs.support.append", true);
        FileSystem fs = null;
        try {
            fs = FileSystem.get(URI.create(hdfsPath+"lowRemoteDetect/lowRemoteDetect.txt"), conf);
            OutputStream output = fs.create(new Path(hdfsPath+"lowRemoteDetect/lowRemoteDetect.txt"));
            String start_format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getStartTime()) + "000000";
            String format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getEndTime()) + "000000";
            Date startTime = new Date();
            Query query = new Query(Criteria.where("pid").gt(start_format).lt(format));
            List<LowRemoteDetect> lowRemoteDetect = mongoTemplate.find(query,LowRemoteDetect.class);
            log.info("低压遥测数据开始写入hdfs");
            for (LowRemoteDetect remoteDetect : lowRemoteDetect) {
                output.write((remoteDetect.getPid()+","+remoteDetect.getRemotDetectId()+","
                        +remoteDetect.getEValue()+"," +remoteDetect.getCreateDate()).getBytes("UTF-8"));
                output.write("\n".getBytes("UTF-8"));
            }
            log.info("低压遥测数据写入hdfs完成");
            Date endTime = new Date();
            systemProcessLogService.addSystemProcessLog("MongoDB_Kudu_接入方式", "ETL", startTime, endTime, lowRemoteDetect.size(), lowRemoteDetect.size(), "Y", "成功");
            log.info("低压遥测数据同步完成"+"即将开始清除mongdb数据");
            try {
                mongoTemplate.findAllAndRemove(query, LowRemoteDetect.class);
            } catch (Exception e) {
                log.error("低压遥测数据清除异常" + e.getMessage());
            }
            log.info("低压遥测数据清除完成");
            fs.close();
            output.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 低压电度信息
     * @param swaggerTimeEntity
     */
    public void moToKueLowRemotePulse(SwaggerTimeEntity swaggerTimeEntity){
        Configuration conf = new Configuration();
        conf.setBoolean("dfs.support.append", true);
        FileSystem fs = null;
        try {
            fs = FileSystem.get(URI.create(hdfsPath+"lowRemotePulse/lowRemotePulse.txt"), conf);
            OutputStream output = fs.create(new Path(hdfsPath+"lowRemotePulse/lowRemotePulse.txt"));
            String start_format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getStartTime()) + "000000";
            String format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getEndTime()) + "000000";
            Date startTime = new Date();
            Query query = new Query(Criteria.where("pid").gt(start_format).lt(format));
            List<LowRemotePulse> lowRemotePulse = mongoTemplate.find(query,LowRemotePulse.class);
            log.info("低压电度数据开始写入hdfs");
            for (LowRemotePulse remotePulse : lowRemotePulse) {
                output.write((remotePulse.getPid()+","+remotePulse.getRemotPulseId()+","
                        +remotePulse.getEValue()+"," +remotePulse.getCreateDate()).getBytes("UTF-8"));
                output.write("\n".getBytes("UTF-8"));
            }
            log.info("低压电度数据写入hdfs完成");
            Date endTime = new Date();
            systemProcessLogService.addSystemProcessLog("MongoDB_Kudu_接入方式", "ETL", startTime, endTime, lowRemotePulse.size(), lowRemotePulse.size(), "Y", "成功");
            log.info("低压电度信息同步完成"+"即将开始清除mongdb数据");
            try {
                mongoTemplate.findAllAndRemove(query, LowRemotePulse.class);
            } catch (Exception e) {
                log.error("低压电度信息清除异常" + e.getMessage());
            }
            log.info("低压电度信息清除完成");
            fs.close();
            output.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 低压遥信信息
     * @param swaggerTimeEntity
     */
    public void moToKuLowRemoteSignal(SwaggerTimeEntity swaggerTimeEntity){
        Configuration conf = new Configuration();
        conf.setBoolean("dfs.support.append", true);
        FileSystem fs = null;
        try {
            fs = FileSystem.get(URI.create(hdfsPath+"lowRemoteSignal/lowRemoteSignal.txt"), conf);
            OutputStream output = fs.create(new Path(hdfsPath+"lowRemoteSignal/lowRemoteSignal.txt"));
            String start_format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getStartTime()) + "000000";
            String format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getEndTime()) + "000000";
            Date startTime = new Date();
            Query query = new Query(Criteria.where("pid").gt(start_format).lt(format));
            List<LowRemoteSignal> lowRemoteSignal = mongoTemplate.find(query,LowRemoteSignal.class);
            log.info("低压遥信数据开始写入hdfs");
            for (LowRemoteSignal remoteSignal : lowRemoteSignal) {
                output.write((remoteSignal.getPid()+","+remoteSignal.getRemoteSignalId()+","
                        +remoteSignal.getEValue()+"," +remoteSignal.getCreateDate()).getBytes("UTF-8"));
                output.write("\n".getBytes("UTF-8"));
            }
            log.info("低压遥信数据写入hdfs完成");
            Date endTime = new Date();
            systemProcessLogService.addSystemProcessLog("MongoDB_Kudu_接入方式", "ETL", startTime, endTime, lowRemoteSignal.size(), lowRemoteSignal.size(), "Y", "成功");
            log.info("低压遥信数据同步完成"+"即将开始清除mongdb数据");
            try {
                mongoTemplate.findAllAndRemove(query, LowRemoteSignal.class);
            } catch (Exception e) {
                log.error("低压遥信数据清除异常" + e.getMessage());
            }
            log.info("低压遥信数据清除完成");
            fs.close();
            output.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 天气数据
     * @param swaggerTimeEntity
     */
    public void moToKuWeather(SwaggerTimeEntity swaggerTimeEntity){
        Configuration conf = new Configuration();
        conf.setBoolean("dfs.support.append", true);
        FileSystem fs = null;
        try {
            fs = FileSystem.get(URI.create(hdfsPath+"weather/weather.txt"), conf);
            OutputStream output = fs.create(new Path(hdfsPath+"weather/weather.txt"));
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            SimpleDateFormat sdf1 = new SimpleDateFormat("yyyyMMddHHmmss");
            Date startTime = new Date();
            List<Weather> weather2 = new ArrayList<Weather>();
            log.info("开始同步气象数据");
            List<Weather> weather = mongoTemplate.findAll(Weather.class);
            for (Weather weather1 : weather) {
                Date parse = sdf.parse(weather1.getCollectionTime());
                String format = sdf1.format(parse);
                weather1.setYear(Integer.valueOf(format.substring(0,4)));
                weather1.setMonth(Integer.valueOf(format.substring(4,6)));
                weather1.setDay(Integer.valueOf(format.substring(6,8)));
                weather1.setInsertTime(sdf.format(swaggerTimeEntity.getEndTime()));
                weather2.add(weather1);
            }
            log.info("气象数据开始写入hdfs");
            for (Weather weather1 : weather2) {
                output.write((weather1.getNewOrOldStation()+","+weather1.getCollectionTime()+","+weather1.getSensorId()
                        +","+weather1.getYear()+","+weather1.getMonth()+","+weather1.getDay()+","+weather1.getSensorName()
                        +","+weather1.getSensorData()+","+weather1.getSensorIsFault()+","+weather1.getInsertTime()).getBytes("UTF-8"));
                output.write("\n".getBytes("UTF-8"));
            }
            log.info("气象数据写入hdfs完成");
            Date endTime = new Date();
            systemProcessLogService.addSystemProcessLog("MongoDB_Kudu_接入方式", "ETL", startTime, endTime, weather2.size(), weather2.size(), "Y", "成功");
            log.info("气象数据同步完成"+"即将开始清除mongdb数据");
            Query query = new Query();
            try {
                mongoTemplate.findAllAndRemove(query, "weather_data");
            } catch (Exception e) {
                log.error("气象数据清除异常" + e.getMessage());
            }
            log.info("气象数据清除完成");
            fs.close();
            output.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 发射机日志信息
     * @param swaggerTimeEntity
     */
    public void moToKuUpTransLog(SwaggerTimeEntity swaggerTimeEntity){
        Configuration conf = new Configuration();
        conf.setBoolean("dfs.support.append", true);
        FileSystem fs = null;
        try {
            fs = FileSystem.get(URI.create(hdfsPath+"upTransLog/upTransLog.txt"), conf);
            OutputStream output = fs.create(new Path(hdfsPath+"upTransLog/upTransLog.txt"));
            String start_format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getStartTime()) + "000000";
            String format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getEndTime()) + "000000";
            Query query = new Query(Criteria.where("pid").gt(start_format).lt(format));
            Date startTime = new Date();
            log.info("开始同步发射机日志数据");
            List<UpTransmitterLog> upTransmitterLog = mongoTemplate.find(query,UpTransmitterLog.class);
            log.info("开始同步发射机日志数据");
            for (UpTransmitterLog transmitterLog : upTransmitterLog) {
                output.write((+new Random().nextInt(1000000000) + ","
                        + transmitterLog.getTransmitterCode() + ","
                        + transmitterLog.getLogTime() + "," + transmitterLog.getLogType()+ ","
                        + transmitterLog.getLogMessage()).getBytes("UTF-8"));
                output.write("\n".getBytes("UTF-8"));
            }
            log.info("发射机日志数据写入hdfs完成");
            Date endTime = new Date();
            systemProcessLogService.addSystemProcessLog("MongoDB_Kudu_接入方式", "ETL", startTime, endTime, upTransmitterLog.size(), upTransmitterLog.size(), "Y", "成功");
            log.info("发射机日志数据同步完成"+"即将开始清除mongdb数据");
            try {
                mongoTemplate.findAllAndRemove(query, UpTransmitterLog.class);
            } catch (Exception e) {
                log.error("发射机日志数据清除异常" + e.getMessage());
            }
            log.info("发射机日志数据清除完成");
            fs.close();
            output.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 发射机连接状态
     * @param swaggerTimeEntity
     */
    public void moToKuBusiTransRunStatus(SwaggerTimeEntity swaggerTimeEntity){
        Configuration conf = new Configuration();
        conf.setBoolean("dfs.support.append", true);
        FileSystem fs = null;
        try {
            fs = FileSystem.get(URI.create(hdfsPath+"busiTransRunStatus/busiTransRunStatus.txt"), conf);
            OutputStream output = fs.create(new Path(hdfsPath+"busiTransRunStatus/busiTransRunStatus.txt"));
            String start_format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getStartTime()) + "000000";
            String format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getEndTime()) + "000000";
            Query query = new Query(Criteria.where("statuses.pid").gt(start_format).lt(format));
            Date startTime = new Date();
            int i = 0;

            log.info("开始同步发射机连接状态信息");
            List<BusiTransRun> findAll = mongoTemplate.find(query,BusiTransRun.class);
            List<BusiTransRunStatus> statusess = new ArrayList<BusiTransRunStatus>();
            for (BusiTransRun busiTransRun : findAll) {
                List<BusiTransRunStatus> statuses = busiTransRun.getStatuses();
                for (BusiTransRunStatus busiTransRunStatus : statuses) {
                    busiTransRunStatus.setTimestamp(busiTransRun.getTimestamp());
                    busiTransRunStatus.setStationCode(busiTransRun.getStationCode());
                    busiTransRunStatus.setMsgType(busiTransRun.getMsgType());
                    busiTransRunStatus.setOperate(busiTransRun.getOperate());
                    statusess.add(busiTransRunStatus);
                }
            }
            log.info("发射机连接状态数据开始写入hdfs");
            for (BusiTransRunStatus busiTransRunStatus : statusess) {
                output.write((busiTransRunStatus.getPid() + "," + busiTransRunStatus.getTransCode() + ","
                        + busiTransRunStatus.getMsgType() + "," + busiTransRunStatus.getOperate() + ","
                        + busiTransRunStatus.getTimestamp() + "," + busiTransRunStatus.getStationCode() + ","
                        + busiTransRunStatus.getConnected() + "," + busiTransRunStatus.getWorkStatus() + ","
                        + busiTransRunStatus.getStatusTime() + "," + busiTransRunStatus.getSource()).getBytes("UTF-8"));
                output.write("\n".getBytes("UTF-8"));
            }
            log.info("发射机连接状态数据写入hdfs完成");
            Date endTime = new Date();
            systemProcessLogService.addSystemProcessLog("MongoDB_Kudu_接入方式", "ETL", startTime, endTime, statusess.size(), statusess.size(), "Y", "成功");
            log.info("发射机连接状态数据同步完成"+"即将开始清除mongdb数据");
            try {
                mongoTemplate.findAllAndRemove(query, BusiTransRun.class);
            } catch (Exception e) {
                log.error("发射机连接状态数据清除异常" + e.getMessage());
            }
            log.info("发射机连接状态数据清除完成");
            fs.close();
            output.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 各系统连接状态信息
     * @param swaggerTimeEntity
     */
    public void moToKuSystemConnStatus(SwaggerTimeEntity swaggerTimeEntity){
        Configuration conf = new Configuration();
        conf.setBoolean("dfs.support.append", true);
        FileSystem fs = null;
        try {
            fs = FileSystem.get(URI.create(hdfsPath+"systemConnStatus/systemConnStatus.txt"), conf);
            OutputStream output = fs.create(new Path(hdfsPath+"systemConnStatus/systemConnStatus.txt"));
            String start_format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getStartTime()) + "000000";
            String format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getEndTime()) + "000000";
            Query query = new Query(Criteria.where("pid").gt(start_format).lt(format));
            Date startTime = new Date();
            int i = 0;
            log.info("开始同步各系统连接状态数据");
            List<SystemConnStatus> findAll = mongoTemplate.find(query,SystemConnStatus.class);
            List<Sysstatuses> sysstatusess = new ArrayList<Sysstatuses>();
            for (SystemConnStatus systemConnStatus : findAll) {
                List<Sysstatuses> sysstatuses = systemConnStatus.getStatuses();
                for (Sysstatuses statuses : sysstatuses) {
                    statuses.setPid(systemConnStatus.getPid());
                    statuses.setOperate(systemConnStatus.getOperate());
                    statuses.setStationCode(systemConnStatus.getStationCode());
                    statuses.setTimestamp(systemConnStatus.getTimestamp());
                    sysstatusess.add(statuses);
                }
            }
            log.info("各系统连接状态数据开始写入hdfs");
            for (Sysstatuses sysstatuses : sysstatusess) {
                output.write((sysstatuses.getPid() + "," + sysstatuses.getSystemCode() + ","
                        + sysstatuses.getOperate() + "," + sysstatuses.getTimestamp() + ","
                        + sysstatuses.getStationCode() + ","+""+ ","
                        + sysstatuses.getStatus()).getBytes("UTF-8"));
                output.write("\n".getBytes("UTF-8"));
            }
            log.info("各系统连接状态数据写入hdfs完成");
            Date endTime = new Date();
            systemProcessLogService.addSystemProcessLog("MongoDB_Kudu_接入方式", "ETL", startTime, endTime, sysstatusess.size(), sysstatusess.size(), "Y", "成功");
            log.info("各系统连接状态数据同步完成"+"即将开始清除mongdb数据");
            try {
                mongoTemplate.findAllAndRemove(query, SystemConnStatus.class);
            } catch (Exception e) {
                log.error("各系统连接状态数据清除异常" + e.getMessage());
            }
            log.info("各系统连接状态数据清除完成");
            fs.close();
            output.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 发射机状态信息秒数据
     * @param swaggerTimeEntity
     */
    public void moToKuUpdetectstatus(SwaggerTimeEntity swaggerTimeEntity){
        Configuration conf = new Configuration();
        conf.setBoolean("dfs.support.append", true);
        FileSystem fs = null;
        try {
            fs = FileSystem.get(URI.create(hdfsPath+"updetectstatus/updetectstatus.txt"), conf);
            OutputStream output = fs.create(new Path(hdfsPath+"updetectstatus/updetectstatus.txt"));
            String start_format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getStartTime()) + "000000";
            String format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getEndTime()) + "000000";
            Query query = new Query(Criteria.where("pid").gt(start_format).lt(format));
            Date startTime = new Date();
            int i = 0;
            log.info("开始同步发射机状态信息秒数据");
            List<Updetectstatus> findAll = mongoTemplate.find(query,Updetectstatus.class);
            List<DectTransmitters> dectToTransmitters = new ArrayList<DectTransmitters>();
            for (Updetectstatus updetectstatus : findAll) {
                List<DectTransmitters> dectTransmitters = updetectstatus.getTransmitters();
                for (DectTransmitters dectTransmitter : dectTransmitters) {
                    dectTransmitter.setPid(updetectstatus.getPid());
                    dectTransmitter.setDetectTime(updetectstatus.getDetectTime());
                    dectTransmitter.setOnPlayProgram(updetectstatus.getOnPlayProgram());
                    dectTransmitter.setOnPlaytransmitter(updetectstatus.getOnPlaytransmitter());
                    dectTransmitter.setTransmitterPowerSum(updetectstatus.getTransmitterPowerSum());
                    dectTransmitter.setTransmitterFrequencySum(updetectstatus.getTransmitterFrequencySum());
                    dectToTransmitters.add(dectTransmitter);
                }
            }
            log.info("发射机状态数据开始写入hdfs");
            for (DectTransmitters dectToTransmitter : dectToTransmitters) {
                output.write((dectToTransmitter.getPid() + "," + dectToTransmitter.getDeptCode() + ","
                        + dectToTransmitter.getTransmitterNo() + "," + dectToTransmitter.getDetectTime() + ","
                        + dectToTransmitter.getTransmitterStatus() + "," + dectToTransmitter.getTransmitterPower() + ","
                        + dectToTransmitter.getTransmitterFrequency() + ","
                        + dectToTransmitter.getTransmitterProgramCode() + "," + dectToTransmitter.getAntennaCode() + ","
                        + dectToTransmitter.getAmplitudeDegree() + "," + dectToTransmitter.getFrequencyDiff() + ","
                        + dectToTransmitter.getOnPlaytransmitter() + "," + dectToTransmitter.getTransmitterPowerSum() + ","
                        + dectToTransmitter.getTransmitterFrequencySum() + "," + dectToTransmitter.getOnPlayProgram()).getBytes("UTF-8"));
                output.write("\n".getBytes("UTF-8"));
            }

            log.info("发射机状态数据写入hdfs完成");
            Date endTime = new Date();
            systemProcessLogService.addSystemProcessLog("MongoDB_Kudu_接入方式", "ETL", startTime, endTime, dectToTransmitters.size(), dectToTransmitters.size(), "Y", "成功");
            log.info("发射机状态数据同步完成"+"即将开始清除mongdb数据");
            try {
                mongoTemplate.findAllAndRemove(query, Updetectstatus.class);
            } catch (Exception e) {
                log.error("发射机状态数据清除异常" + e.getMessage());
            }
            log.info("发射机状态数据清除完成");
            fs.close();
            output.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 综合分析与报警信息
     * @param swaggerTimeEntity
     */
    public void moToKuBusiAlarmInfo(SwaggerTimeEntity swaggerTimeEntity){
        Configuration conf = new Configuration();
        conf.setBoolean("dfs.support.append", true);
        FileSystem fs = null;
        try {
            fs = FileSystem.get(URI.create(hdfsPath+"busiAlarmInfo/busiAlarmInfo.txt"), conf);
            OutputStream output = fs.create(new Path(hdfsPath+"busiAlarmInfo/busiAlarmInfo.txt"));
            String start_format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getStartTime()) + "000000";
            String format = new SimpleDateFormat("yyyyMMdd").format(swaggerTimeEntity.getEndTime()) + "000000";
            Query query = new Query(Criteria.where("pid").gt(start_format).lt(format));
            Date startTime = new Date();
            int i = 0;
            log.info("开始同步综合分析与报警数据");
            List<BusiAlarmInfo> busiAlarmInfo = mongoTemplate.find(query,BusiAlarmInfo.class);
            for (BusiAlarmInfo alarmInfo : busiAlarmInfo) {
                output.write((new Random().nextInt(1000000000)+","+""+ "," + alarmInfo.getOperate() + ","
                        + alarmInfo.getTimestamp() + "," + alarmInfo.getBAIid() + ","
                        + alarmInfo.getStationCode() + ",'"
                        + alarmInfo.getTransCode() + "," + alarmInfo.getAlarmLevel() + ","
                        + alarmInfo.getDescription() + "," + alarmInfo.getStartTime() + ","
                        + alarmInfo.getEndTime()).getBytes("UTF-8"));
                output.write("\n".getBytes("UTF-8"));
            }
            log.info("综合分析与报警数据写入hdfs完成");
            Date endTime = new Date();
            systemProcessLogService.addSystemProcessLog("MongoDB_Kudu_接入方式", "ETL", startTime, endTime, busiAlarmInfo.size(), busiAlarmInfo.size(), "Y", "成功");
            log.info("综合分析与报警数据同步完成"+"即将开始清除mongdb数据");
            try {
                mongoTemplate.findAllAndRemove(query, BusiAlarmInfo.class);
            } catch (Exception e) {
                log.error("综合分析与报警数据清除异常" + e.getMessage());
            }
            log.info("综合分析与报警数据清除完成");
            fs.close();
            output.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 警告信息
     * @param swaggerTimeEntity
     */
    public void moToKuAmbiTranAlarm(SwaggerTimeEntity swaggerTimeEntity){
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Configuration conf = new Configuration();
        conf.setBoolean("dfs.support.append", true);
        FileSystem fs = null;
        try {
            fs = FileSystem.get(URI.create(hdfsPath+"ambiTranAlarm/ambiTranAlarm.txt"), conf);
            OutputStream output = fs.create(new Path(hdfsPath+"ambiTranAlarm/ambiTranAlarm.txt"));
            String lastTime = DateUtil.getLastTime(2);
            Query query = new Query(Criteria.where("alarmTime").lt(lastTime));
            Date startTime = new Date();
            int i = 0;
            log.info("开始同步告警数据");
            List<AmbiTranAlarm> ambiTranAlarm = mongoTemplate.find(query,AmbiTranAlarm.class);
            for (AmbiTranAlarm tranAlarm : ambiTranAlarm) {
                output.write((new Random().nextInt(1000000000)+","+ tranAlarm.getAlarmType() + ","
                        + tranAlarm.getAlarmTime() + ","+ tranAlarm.getAlarmEndTime()+","
                        + tranAlarm.getAlarmData().replaceAll(",", "")+ ","
                        + sdf.format(new Date())).getBytes("UTF-8"));
                output.write("\n".getBytes("UTF-8"));
            }
            log.info("告警数据写入hdfs完成");
            Date endTime = new Date();
            systemProcessLogService.addSystemProcessLog("MongoDB_Kudu_接入方式", "ETL", startTime, endTime, ambiTranAlarm.size(), ambiTranAlarm.size(), "Y", "成功");
            log.info("告警数据同步完成"+"即将开始清除mongdb数据");
            try {
                mongoTemplate.findAllAndRemove(query, AmbiTranAlarm.class);
            } catch (Exception e) {
                log.error("告警数据清除异常" + e.getMessage());
            }
            log.info("告警数据清除完成");
            fs.close();
            output.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }


    /**
     * 发射机秒数据
     * @param secondByDept
     * @param collectionName
     */
    public void moToKuUpTransmitterSecondData(String secondByDept,String collectionName){
        log.info("开始同步");
        SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Configuration conf = new Configuration();
        conf.setBoolean("dfs.support.append", true);
        FileSystem fs = null;
        try {
            fs = FileSystem.get(URI.create(hdfsPath+secondByDept), conf);
            OutputStream output = fs.create(new Path(hdfsPath+secondByDept));
            String transFormat = new SimpleDateFormat("yyyyMMdd").format(new Date()) + "000000";
            Query query = new Query(Criteria.where("transmitters.pid").lt(transFormat));
            List<UpTransmitterSecondData> findAll = mongoTemplate.find(query,UpTransmitterSecondData.class,collectionName);
            List<Transmitters> transmitterss = new ArrayList<Transmitters>();
            for (UpTransmitterSecondData upTransmitterSecondData : findAll) {
                List<Transmitters> transmitters = upTransmitterSecondData.getTransmitters();
                for (Transmitters transmitter : transmitters) {
                    if(transmitter.getPid().length()>10){
                        transmitter.setDataLength(upTransmitterSecondData.getDataLength());
                        transmitter.setDeptCode(upTransmitterSecondData.getDeptCode());
                        transmitter.setTransmitterCount(upTransmitterSecondData.getTransmitterCount());
                        transmitterss.add(transmitter);
                    }
                }
            }
            log.info("发射机秒数据开始写入hdfs,写入条数为"+transmitterss.size());
            for (Transmitters transmitters : transmitterss) {
                String format = formatter.format(formatter.parse(transmitters.getGetTime()));
                output.write((new Random().nextInt(1000000000)+","
                        + transmitters.getPid() + "," + transmitters.getDeptCode() + ","
                        + transmitters.getTransmitterCode() + "," + transmitters.getDataLength() + ","
                        + transmitters.getTransmitterCount() + "," + format + ","
                        + transmitters.getSystemStatus() + "," + transmitters.getTransmitterStatus()+ ","
                        + transmitters.getSwitchCount() + "," + transmitters.getSwitch_1() + ","
                        + transmitters.getSwitch_2() + "," + transmitters.getSwitch_3() + ","
                        + transmitters.getSwitch_4() + "," + transmitters.getSwitch_5() + ","
                        + transmitters.getSwitch_6() + "," + transmitters.getSwitch_7() + ","
                        + transmitters.getSwitch_8() + "," + transmitters.getSwitch_9() + ","
                        + transmitters.getSwitch_10() + "," + transmitters.getSwitch_11() + ","
                        + transmitters.getSwitch_12() + "," + transmitters.getSwitch_13() + ","
                        + transmitters.getSwitch_14() + "," + transmitters.getSwitch_15() + ","
                        + transmitters.getSwitch_16() + "," + transmitters.getSwitch_17() + ","
                        + transmitters.getSwitch_18() + "," + transmitters.getSwitch_19() + ","
                        + transmitters.getSwitch_20() + "," + transmitters.getSwitch_21() + ","
                        + transmitters.getSwitch_22() + "," + transmitters.getSwitch_23() + ","
                        + transmitters.getSwitch_24() + "," + transmitters.getSwitch_25() + ","
                        + transmitters.getSwitch_26() + "," + transmitters.getSwitch_27() + ","
                        + transmitters.getSwitch_28() + "," + transmitters.getSwitch_29() + ","
                        + transmitters.getSwitch_30() + "," + transmitters.getSwitch_31() + ","
                        + transmitters.getSwitch_32() + "," + transmitters.getSwitch_33() + ","
                        + transmitters.getSwitch_34() + "," + transmitters.getSwitch_35() + ","
                        + transmitters.getSwitch_36() + "," + transmitters.getSwitch_37() + ","
                        + transmitters.getSwitch_38() + "," + transmitters.getSwitch_39() + ","
                        + transmitters.getSwitch_40() + "," + transmitters.getSimulationCount() + ","
                        + transmitters.getSimulation_1()+ ","
                        + transmitters.getSimulation_2() + "," + transmitters.getSimulation_3() + ","
                        + transmitters.getSimulation_4() + "," + transmitters.getSimulation_5() + ","
                        + transmitters.getSimulation_6() + "," + transmitters.getSimulation_7() + ","
                        + transmitters.getSimulation_8() + "," + transmitters.getSimulation_9() + ","
                        + transmitters.getSimulation_10() + "," + transmitters.getSimulation_11() + ","
                        + transmitters.getSimulation_12() + "," + transmitters.getSimulation_13() + ","
                        + transmitters.getSimulation_14() + "," + transmitters.getSimulation_15() + ","
                        + transmitters.getSimulation_16() + "," + transmitters.getSimulation_17() + ","
                        + transmitters.getSimulation_18() + "," + transmitters.getSimulation_19() + ","
                        + transmitters.getSimulation_20() + "," + transmitters.getSimulation_21() + ","
                        + transmitters.getSimulation_22() + "," + transmitters.getSimulation_23() + ","
                        + transmitters.getSimulation_24() + "," + transmitters.getSimulation_25() + ","
                        + transmitters.getSimulation_26() + "," + transmitters.getSimulation_27() + ","
                        + transmitters.getSimulation_28() + "," + transmitters.getSimulation_29() + ","
                        + transmitters.getSimulation_30()).getBytes("UTF-8"));
                output.write("\n".getBytes("UTF-8"));
            }
            log.info("发射机秒数据写入hdfs完成,即将开始清除mongdb数据");
            try {
                mongoTemplate.findAllAndRemove(query, UpTransmitterSecondData.class,collectionName);
            } catch (Exception e) {
                log.error("发射机秒数据清除异常" + e.getMessage());
            }
            log.info("发射机秒数据清除完成");
            fs.close();
            output.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}
